package com.mgface.kafka.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author wanyuxiang
 * @version 1.0
 * @project kafkademo
 * @create 2019-01-18 10:29
 **/
@Slf4j
public class ProducerMain {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties ps = new Properties();
        ps.put("bootstrap.servers", "localhost:9092");
        ps.put("key.serializer", StringSerializer.class.getName());
        ps.put("value.serializer", StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<>(ps);
        for (int i = 0; i < 50; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("topicC", "admin"+i, "hello,world");
            Future<RecordMetadata> metadata = producer.send(record, (meta, e) -> log.info("回调方法返回:{}", meta));
            log.info("返回的结果为:{}", metadata.get());
        }
    }
}
